// Copyright 2014 The Go Circuit Project
// Use of this source code is governed by the license for
// The Go Circuit Project, found in the LICENSE file.
//
// Authors:
//   2014 Chris Monson <chris@gocircuit.org>

package tokenlock

import (
	"errors"
	"runtime"
	"sync"

	"circuit/kit/rh"
)

var (
	ErrReset        = errors.New("Resource: reset while waiting")
	ErrInterrupted  = errors.New("Resource: wait interrupted")
	ErrInvalidToken = errors.New("Token: no longer valid")
)

// Token is the handed back when Acquire succeeds. So long as it IsValid, the
// holder has exclusive access to the resource guarded by the Lock.
type Token interface {
	// IsValid returns true if the token is valid. It could become invalid at
	// any time, so usually you will want to use DoIfValid to ensure that it
	// remains valid for the duration of some operation.
	IsValid() bool

	// This atomically performs do only if the token is valid. Possible return values:
	// - nil: The token is held and the operation completed.
	// - ErrInvalidToken: the token is not valid and the operation was not performed.
	DoIfValid(do func()) error

	// Release releases the token to the trackable mutex that assigned it. Immediately returns
	// a channel on which an error can be received:
	// - nil after the completion of a successful release
	// - ErrReset after the completion of a release superceded by a reset
	// - ErrInvalidToken immediately if the token is already invalid
	Release() <-chan error
}

type Lock interface {
	// Acquire attempts to acquire a lock token. Sending any value on the intr
	// channel will cause the wait to be abandoned. The following errors are
	// possible:
	// - nil: token successfully acquired
	// - ErrReset: the Lock was reset and all waiters were forcibly dumped.
	// - ErrInterrupted: the intr channel received a value (any value).
	// Note that a token is returned every time, even if there is an error. The
	// token will be invalid, but this can help with idiomatic use of
	// 		defer token.Release()
	// which is harmless on an invalid token, and can add safety.
	Acquire(interrupt rh.Intr) (Token, error)

	// Stat returns the number of tokens held (0 or 1) and the number of
	// blocked waiters.
	Stat() (held, waiting int)

	// Reset causes all waiters to stop waiting and invalidates any held token.
	// It returns a channel that callers can listen on to wait until the Reset
	// is finished.
	Reset() <-chan error

	// WaitIdle blocks until there are no tokens held and no waiting processes.
	WaitIdle()
}

// tokenResourceLock implements Lock.
type tokenResourceLock struct {
	sync.Mutex
	lockChan   chan Token
	resetChan  chan chan error // send a channel for communication of completion
	numWaiting int
	held       *lToken
	condIdle   *sync.Cond
}

// New creates a new Lock to guard a resource.
func New() Lock {
	m := &tokenResourceLock{
		lockChan:  make(chan Token),
		resetChan: make(chan chan error),
	}
	m.condIdle = sync.NewCond(&sync.Mutex{})
	go m.manageLocks()
	return m
}

// Reset signals all current waiters to abandon the wait and invalidates the
// current holder's token. The Lock quickly ends up in its initial state
// after this.
// Returns a channel that will receive a single value when the reset is complete.
func (m *tokenResourceLock) Reset() <-chan error {
	defer un(lock(m))
	// Make the current holder lose its hold immediately, but retain our
	// ability to send a release ourselves.
	// Signals all waiters and terminates lock token for holder.
	ch := make(chan error, 1) // don't block to notify receiver, who may not care
	m.resetChan <- ch
	return ch
}

// Stat returns whether this lock is held and how many are waiting for it.
func (m *tokenResourceLock) Stat() (held, waiting int) {
	defer un(lock(m))
	if m.held != nil {
		held = 1
	}
	return held, m.numWaiting
}

// Acquire attempts to acquire a lock token. Sending any value on the intr
// channel will cause the wait to be abandoned. The following errors are
// possible:
// - nil: token successfully acquired
// - ErrReset: the Lock was reset and all waiters were forcibly dumped.
// - ErrInterrupted: the intr channel received a value (any value).
// Note that a token is returned every time, even if there is an error. The
// token will be invalid, but this can help with idiomatic use of
// 		defer token.Release()
// which is harmless on an invalid token, and can add safety.
func (m *tokenResourceLock) Acquire(intr rh.Intr) (Token, error) {
	lch := m.prepareOneWaiter()
	select {
	case token, ok := <-lch:
		if !ok {
			// The channel was closed and is not accepting any more locks.
			// The wait count is already zero by this point.
			return newInvalidToken(), ErrReset
		}
		return token, nil
	case <-intr:
		m.abandonOneWaiter()
		// Interrupted from upstream.
		return newInvalidToken(), ErrInterrupted
	}
}

// WaitIdle blocks until the token producer has no outstanding tokens and no
// waiting processes. It does not acquire a token.
func (m *tokenResourceLock) WaitIdle() {
	defer un(lock(m.condIdle.L))
	for nh, nw := m.Stat(); nh != 0 || nw != 0; nh, nw = m.Stat() {
		m.condIdle.Wait()
	}
}

// lock of the "defer un(lock(l))" pattern.
func lock(m sync.Locker) sync.Locker {
	m.Lock()
	return m
}

// un of the "defer un(lock(l))" pattern.
func un(m sync.Locker) {
	m.Unlock()
}

func (m *tokenResourceLock) prepareOneWaiter() <-chan Token {
	defer un(lock(m))
	m.numWaiting++
	return m.lockChan
}

func (m *tokenResourceLock) abandonOneWaiter() {
	defer un(lock(m))
	defer m.condIdle.Broadcast()
	if m.numWaiting == 0 {
		panic("No waits to abandon: can't happen")
	}
	m.numWaiting--
}

func (m *tokenResourceLock) admitNextWaiter(token *lToken) {
	defer un(lock(m))
	defer m.condIdle.Broadcast()
	if m.held != nil {
		panic("Admitting next when lock already held. Can't happen.")
	}
	if m.numWaiting == 0 {
		panic("Nobody waiting, but someone just got the token. Can't happen.")
	}
	m.held = token
	m.numWaiting--
}

func (m *tokenResourceLock) forgetHolder() {
	defer un(lock(m))
	defer m.condIdle.Broadcast()
	if m.held == nil {
		panic("Releasing holder, but there is no holder. Can't happen.")
	}
	m.held = nil
}

func (m *tokenResourceLock) massRelease() {
	defer un(lock(m))
	defer m.condIdle.Broadcast()
	if m.held != nil {
		m.held.Release()
	}
	m.held = nil
	close(m.lockChan)
	m.lockChan = make(chan Token)
	m.numWaiting = 0
}

func (m *tokenResourceLock) manageLocks() {
	// uch is the unlock channel
	var uch chan chan error
	token := newValidToken()
	lch := m.lockChan

	for {
		select {
		case lch <- token:
			// If we can push a token into the lock channel, then someone is
			// waiting for it and we don't have too many out. Otherwise, lch is
			// nil, so we never select this case.
			lch = nil // no more locks until unlocked
			m.admitNextWaiter(token)
			uch = token.unlock // allow new unlocks
			token = nil        // next lock token must be new
		case done := <-uch:
			// The unlock channel receives a channel to which we can send an
			// error code. A token release triggers this case. We send a nil
			// error back to indicate that everything worked.
			uch = nil // no more unlocks until locked again
			m.forgetHolder()
			token = newValidToken()
			lch = m.lockChan
			done <- nil
		case done := <-m.resetChan:
			// Any tokens that are outstanding might still try to send on the
			// unlock channel, then block on "done". Here we ensure that they
			// get a chance to unblock.
			//
			// Note that we are guaranteed to receive something on "late"
			// because massRelease calls held.Release(). Therefore, the token
			// holder will either beat us to the reset and we'll send a value on
			// "done", or we will beat them to it and they will find that the
			// token is already invalid when they try to release.
			if uch != nil {
				go func(late chan chan error) {
					done := <-late
					if done != nil {
						done <- ErrReset
					}
				}(uch)
				uch = nil
			}
			m.massRelease()
			token = newValidToken()
			lch = m.lockChan
			if done != nil {
				done <- nil
				close(done)
			}
		}
	}
}

type lToken struct {
	sync.Mutex
	unlock chan chan error // send a "done" channel to release
}

func newValidToken() *lToken {
	t := &lToken{unlock: make(chan chan error, 1)}
	runtime.SetFinalizer(t, func(x *lToken) {
		x.Release()
	})
	return t
}

func newInvalidToken() *lToken {
	return &lToken{}
}

// IsValid returns true iff the token is held and the underlying Lock has
// not been reset.
func (t *lToken) IsValid() bool {
	defer un(lock(t))
	return t.unlock != nil
}

// DoIfValid atomically executes 'do' and returns true if the token is valid,
// and returns false otherwise.
func (t *lToken) DoIfValid(do func()) (err error) {
	defer un(lock(t))
	if t.unlock == nil {
		return ErrInvalidToken
	}
	do()
	return nil
}

// Release invalidates this token and signals the Lock that someone else
// can obtain one. It immediately returns a notification channel. If the token
// is invalid, the channel will come back closed. Otherwise it will receive a
// value when the release has been recorded.
func (t *lToken) Release() <-chan error {
	defer un(lock(t))
	done := make(chan error, 1) // don't block the sender: this may never be received.
	if t.unlock == nil {
		done <- ErrInvalidToken
		return done
	}
	t.unlock <- done // signal that we're done with this token
	close(t.unlock)  // unlock channels are only good once
	t.unlock = nil   // invalidate this token
	return done
}
